package canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;


public class SimpleCanalClientExample {

    private final static String ip = "192.168.10.168";
    private final static int port = 11111;

    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors.
                newSingleConnector(new InetSocketAddress(ip, port), "example", "", "");
        int batchSize = 1000;
         try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
             while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                 try {
                     if (batchId == -1 || size == 0) {

                     } else {
                         // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                         printEntry(message.getEntries());
                     }
                     connector.ack(batchId); // 提交确认
                 }catch (Exception e){
                     connector.rollback(batchId); // 处理失败, 回滚数据
                 }

            }

         } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            //操作类型
            EventType operationType = rowChage.getEventType();
            //数据库名
            String databaseName = entry.getHeader().getSchemaName();
            //表名
            String tableName = entry.getHeader().getTableName();

            //过滤操作类型，只记录，增，改
            if(operationType.toString().equals("DELETE")) continue;
            //过滤数据库，只记录test数据库
            if(!entry.getHeader().getSchemaName().equals("test")) continue;
            StringBuffer sql = new StringBuffer();

            //遍历内容
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (operationType == EventType.INSERT) {
                     sql.append(" insert into "+tableName +" VALUES ( ");
                    List<Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (Column column : afterColumnsList) {
                        String value = column.getValue();
                        if("".equals(value)||value.length()==0){
                            value=null;
                            sql.append( value +",");

                        }else {
                            sql.append("'"+ value +"',");
                        }
                     }
                    sql.deleteCharAt(sql.length()-1);
                    sql.append(")");

                } else if(operationType == EventType.UPDATE){
                    sql.append("update "+ tableName+"  set ");
                    List<Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (Column column : afterColumnsList) {

                        //字段有更新
                        if(column.getUpdated()){
                            String s = column.getName()+" = '" +column.getValue() + "' ,";
                            sql.append(s);
                        }
                    }
                    sql.deleteCharAt(sql.length()-1);
                    Column columnId = afterColumnsList.get(0);
                    sql.append(" where   "+columnId.getName()+"='"+columnId.getValue()+"'");
                 }
            }
             String s = "sql语句："+sql.toString();
            System.out.println(s);
        }
    }

    


}